package com.tengju.data.infrastructure.kafka.consumer;

import com.tengju.data.domain.model.message.FlatMessage;
import com.tengju.data.domain.shared.JsonUtil;
import com.tengju.data.infrastructure.persistence.mybatis.mapper.data.custom.AbmBinlogMapper;
import com.tengju.data.infrastructure.shared.InfraException;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.HashSet;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Optional;
import java.util.Set;
import java.util.regex.Pattern;

@Component
@Slf4j
@ConditionalOnProperty(prefix = "binlog.consumer", name = "switch", havingValue = "true")
@RequiredArgsConstructor(onConstructor_ = {@Autowired})
public class AbmBinlogConsumer {

    private final AbmBinlogMapper abmBinlogMapper;

    private static final Pattern SPECIAL_REGEX_CHARS = Pattern.compile("[{}()\\[\\].+*?^$\\\\|\"']");


    @KafkaListener(
            topics = "#{'${kafka.listener.topics.abm.binlog}'.split(',')}",
            containerFactory = "abmBinlogKafkaListenerContainerFactory",
            errorHandler = "binlogConsumerListenerErrorHandler"
    )
    public void listener(ConsumerRecord<String, String> record, Acknowledgment ack)
            throws InfraException {
        Optional<String> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isEmpty()) {
            throw new InfraException("received kafkaMessage empty");
        }
        String binlog = kafkaMessage.get();
        log.info("receive abm binlog: {}", binlog);
        FlatMessage flatMessage = JsonUtil.toObject(binlog, FlatMessage.class);
        if (flatMessage == null) {
            throw new InfraException("binlog toObject error");
        }
        String type = flatMessage.getType();
        if (StringUtils.isBlank(type)) {
            throw new InfraException("binlog type is empty");
        }
        String sql;
        switch (type) {
            case "INSERT":
                sql = generateInsertSQL(flatMessage);
                break;
            case "UPDATE":
                sql = generateUpdateSQL(flatMessage);
                break;
            case "DELETE":
                sql = generateDeleteSQL(flatMessage);
                break;
            case "ALTER":
                sql = flatMessage.getSql();
                break;
            default:
                throw new InfraException("type not support");
        }
        if (StringUtils.isNotBlank(sql)) {
            abmBinlogMapper.execute(sql);
        }
        ack.acknowledge();
    }

    private String generateInsertSQL(FlatMessage flatMessage) {
        StringBuilder sql = new StringBuilder();
        sql.append(String.format("REPLACE INTO %s SET ", flatMessage.getTable()));
        int i = 0;
        for (Entry<String, String> entry : flatMessage.getData().entrySet()) {
            String value = entry.getValue();
            if (value != null) {
                sql.append(String.format("%s = '%s'", entry.getKey(), resolveValue(entry.getValue())));
            } else {
                sql.append(String.format("%s = %s", entry.getKey(), null));
            }
            i++;
            if (i == flatMessage.getData().entrySet().size()) {
                sql.append(";");
            } else {
                sql.append(",");
            }
        }
        return sql.toString();
    }


    /**
     * 处理sql的特殊字符
     *
     * @param
     * @return
     */
    public String resolveValue(String str) {
        if (StringUtils.isEmpty(str)) {
            return str;
        }
        return SPECIAL_REGEX_CHARS.matcher(str).replaceAll("\\\\$0");
    }

    private String generateDeleteSQL(FlatMessage flatMessage) {
        StringBuilder sql = new StringBuilder();
        String primaryKeyName = flatMessage.getPkNames().get(0);
        if (StringUtils.isBlank(primaryKeyName)) {
            throw new InfraException("primaryKeyName is empty");
        }
        sql.append(String.format("DELETE FROM %s ", flatMessage.getTable()));
        sql.append(String.format(" where %s = '%s'", primaryKeyName,
                flatMessage.getData().get(primaryKeyName)));
        return sql.toString();
    }

    private String generateUpdateSQL(FlatMessage flatMessage) {
        StringBuilder sql = new StringBuilder();
        String primaryKeyName = flatMessage.getPkNames().get(0);
        if (StringUtils.isBlank(primaryKeyName)) {
            throw new InfraException("primaryKey is empty");
        }
        sql.append(String.format("UPDATE %s SET ", flatMessage.getTable()));
        Map<String, String> data = flatMessage.getData();
        Map<String, String> old = flatMessage.getOld();
        Set<String> set = new HashSet<>();
        // 从null修改为值
        generateUpdateField(primaryKeyName, data, old, set);
        for (String s : set) {
            sql.append(s);
        }
        if (sql.toString().endsWith(",")) {
            sql.deleteCharAt(sql.toString().length() - 1);
        }
        sql.append(String.format(" where %s = '%s'", primaryKeyName,
                resolveValue(flatMessage.getData().get(primaryKeyName))));
        return sql.toString();
    }

    private void generateUpdateField(String primaryKeyName, Map<String, String> data, Map<String, String> old, Set<String> set) {
        for (Entry<String, String> entry : old.entrySet()) {
            if (primaryKeyName.equalsIgnoreCase(entry.getKey())) {
                continue;
            }
            String value = data.get(entry.getKey());
            if (value != null) {
                set.add(String.format("%s = '%s',", entry.getKey(), resolveValue(value)));
            } else {
                set.add(String.format("%s = %s,", entry.getKey(), null));
            }
        }
    }
}
